#include "dbenv.hxx"
#include "datastore.hxx"
#include "util/common.hxx"
#include "dbt.hxx"
#include <iostream>
#include <sstream>
#include <assert.h>


int n = 500000;
int start, end;

std::string mymapfunc(Dbt& key) {

    return "evenpartition";
}    

std::string createValue(const std::string& key) {

    std::string v = key;
    std::string valuestr(128, 'y');

    return v + valuestr;
}

int sequentialWrite(DbEnv* env, DataStore* store) {

    DbStatus ret;

    for(int i = start; i <= end; ++i)
    {
        std::ostringstream oss; oss<<i; 
        std::string key = oss.str();
        Dbt keydbt(key); 
        std::string valuestr = createValue(key);
        Dbt value(valuestr);
        ret = store->put(keydbt, value);
        if(i%1000 == 0)
        //std::cout<<i<<"\n";
        if(!ret.isok())  {std::cout<<"put failed"<<ret.msg(); return -1;}
/*
        if(i > 0 && i % 20 == 0) 
        {
           ret =  env->sync();
           if(!ret.isok())  {std::cout<<"put sync"<<ret.msg(); return -1;}
        }
*/
    }


    return 0;
}

int sequentialread(DbEnv* env, DataStore* store) {

    for(int i = start; i <= end; ++i) {
    
        std::ostringstream oss; oss<<i;
        std::string key = oss.str();
        Dbt keydbt(key), valuedbt;
        DbStatus ret = store->get(keydbt, valuedbt);
        if(!ret.isok())  {std::cout<<"get failed"<<ret.msg(); return -1;}
        
        std::string value0 = createValue(key);
        if(value0 != valuedbt.toString()) {
            std::cout<<i<<" not matched: "<<std::endl;
        }
    }
}

int main(int argc, char** argv) {

    start = utility::atoi(argv[2]);
    end = utility::atoi(argv[3]);

    std::string logmsgFile = argv[1];
    logmsgFile = logmsgFile + "/" + "mydb.msg";
    uint32_t bufferSize = 5*1024*1024;

    DbEnv* env = new DbEnv(argv[1], argv[1], logmsgFile, bufferSize);
    
    uint32_t flag = LOG_NO_DATA;
    DbStatus ret = env->open(flag);
    if(!ret.isok()) {std::cout<<"env cannot be opened"<<ret.msg(); return -1;}


   DataStore* store = new DataStore(*env);
   const std::string dsName("mystore");
   uint32_t cacheSize = 1000000; 
   uint32_t indexArrayCapacity = 64*1024; 
   uint32_t segCapacity = 32*1024*1024; 
   MAPPINGFUNC* mapFunc = mymapfunc;
   ret = store->open(dsName, cacheSize, indexArrayCapacity, segCapacity,
           mapFunc, 0);
   if(!ret.isok())  {std::cout<<"open"<<ret.msg(); return -1;}

   ret = store->addPartition("oddpartition");
   if(!ret.isok())  {std::cout<<"add partition"<<ret.msg(); return -1;}
   ret = store->addPartition("evenpartition");
   if(!ret.isok())  {std::cout<<"add partition"<<ret.msg(); return -1;}

   ret = env->start(); 
    if(!ret.isok())  {std::cout<<"env start"<<ret.msg(); return -1;}

   if(argv[2][0] == 'w') {
       std::cout<<"writing mode\n";
       env->startCompact();
       sequentialWrite(env, store);
   }
   else if(argv[2][0] == 'r') {
   
        std::cout<<"reading mode\n";
        sequentialread(env, store);
   }

   env->sync();
//   env->flushlog();

   //store->flush();

   delete store;
   delete env;
}

